package com.paradogs.framework.core.utils;

import com.paradogs.framework.core.annoations.request.PRMsgMappingHolder;
import com.paradogs.framework.core.events.RPCRequestEvent;
import com.paradogs.framework.core.exceptions.MessageException;
import com.paradogs.framework.core.msg.PRMsgContextHolder;
import com.paradogs.framework.core.msg.route.MsgMapping;
import com.paradogs.framework.core.netty.PRConnHolder;
import com.paradogs.framework.core.netty.server.PRConnection;
import com.paradogs.framework.core.netty.server.ServerProperties;
import com.paradogs.framework.core.msg.PRMsg;
import com.paradogs.framework.core.msg.ByteData;
import com.paradogs.framework.core.msg.MsgHeader;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.event.EventListener;
import org.springframework.stereotype.Component;
import sun.misc.GC;

import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.util.concurrent.CompletableFuture;
import java.util.stream.Stream;


/**
 * @author: yumi
 * @date: 2023/7/27  10:33
 * @Description: TODO
 */
@Slf4j
@Component
public class MsgUtils {

    private static ServerProperties serverProperties;

    @Autowired
    public void setServerProperties(ServerProperties serverProperties) {
        MsgUtils.serverProperties = serverProperties;
    }

    /**
     * 发起 rpc 请求
     * @param event
     * 待优化：先发送再监听可能会赶不上，概率较小
     */
    @EventListener(RPCRequestEvent.class)
    public void onRPCRequestEvent(RPCRequestEvent event) {
        // 发起请求
        MsgHeader msgHeader = PRMsgContextHolder.getMsgHeader();

        PRMsg msg = send(event.getRoute(), event.getData());
        // 开启响应监听
        if (event.getReturnClass() == null || event.getReturnClass().equals(Void.TYPE)) {
            return;
        }
        EventUtils.on4Once(event.getRoute(), msg.getMsgHeader().getId(), event.getReturnClass(), event.getPromise());
    }

    /**
     * 根据 route 适配执行对应的 Handler
     * @param msg
     * @throws InvocationTargetException
     * @throws IllegalAccessException
     */
    public static void adapterHandle(ByteData msg) throws InvocationTargetException, IllegalAccessException {

        MsgHeader header = msg.getMsgHeader();

        // 获取映射
        MsgMapping mapping = PRMsgMappingHolder.getMsgMapping(RouteUtils.getPath(header.getRoute()));
        if (mapping == null) {
            log.warn("Not found Mapping Method, route: {}", RouteUtils.getPath(header.getRoute()));
            return;
        }
        Class<?> msgClass = PRMsgMappingHolder.getMsgMappingClass(RouteUtils.getPath(header.getRoute()));

        /**
         * 执行对应 Handler 方法
         */
        Method method = mapping.getMethod();
        Object pdgController = SpringUtils.getBean(method.getDeclaringClass());

        if (pdgController == null) {
            log.error("Class not managed by Spring: {}", method.getDeclaringClass());
            return;
        }

        /**
         * 待优化：这里可以像 SpringMVC 一样，根据方法参数注入相关变量
         */
        if (mapping.isAsync()) {
            /**
             * 异步执行
             */
            final PRMsgContextHolder.Snapshot mainThreadContextSnapshot = PRMsgContextHolder.getSnapshot();
            PRAsync.run(() -> {
                try {
                    // 将主线程上下文设置到异步线程上下文
                    PRMsgContextHolder.setContext(mainThreadContextSnapshot);
                    // 执行方法
                    Object result = method.invoke(
                            pdgController,
                            Stream.of(method.getParameterTypes())
                                    .map(type -> {
                                        // 转换为参数类型传参
                                        if (type.isAssignableFrom(msgClass)) {
                                            if (msgClass.isAssignableFrom(ByteData.class)) {
                                                return msg;
                                            } else {
                                                // 传入 data 参数
                                                return ProtoUtils.decode(msg.getDataBytes(), msgClass);
                                            }
                                        }
                                        return null;
                                    })
                                    .toArray()
                    );
                    /**
                     * 有返回结果时，自动将返回结果作为 GC 消息发送
                     */
                    if (method.getReturnType().equals(Void.TYPE)) {
                        return;
                    }
                    reply(header.getSendServer(), result, msg.getMsgHeader().getId(), PRMsgContextHolder.getChannel());
                } catch (Exception e) {
                    throw new MessageException(e);
                } finally {
                    // 释放上下文，切面可能会受影响取不到？ 待优化
                    PRMsgContextHolder.releaseContext();
                }
            });
        } else {
            /**
             * 同步执行（正常执行）
             */
            Object result = method.invoke(
                    pdgController,
                    Stream.of(method.getParameterTypes())
                            .map(type -> {
                                // 转换为参数类型传参
                                if (type.isAssignableFrom(msgClass)) {
                                    if (msgClass.isAssignableFrom(ByteData.class)) {
                                        return msg;
                                    } else {
                                        // 传入 data 参数
                                        return ProtoUtils.decode(msg.getDataBytes(), msgClass);
                                    }
                                }
                                return null;
                            })
                            .toArray()
            );
            /**
             * 有返回结果时，自动将返回结果作为 GC 消息发送
             */
            if (method.getReturnType().equals(Void.TYPE)) {
                return;
            }
            reply(header.getSendServer(), result, msg.getMsgHeader().getId(), PRMsgContextHolder.getChannel());
        }

    }

    /**
     * 服务器之间发送消息
     * @param route 路由
     * @param data 数据
     * @return
     */
    public static <T> PRMsg send(String route, T data, Channel channel) {
        PRMsg msg = new PRMsg(route, data);
        msg.getMsgHeader().setPId(PRMsgContextHolder.getPlayerId());
        channel.writeAndFlush(msg);
        log.debug(">>>>> {}, {}, {}", route, data, msg.getMsgHeader().getId());
        return msg;
    }
    protected static <T> PRMsg send(String route, T data) {
        Channel channel = getChannel(RouteUtils.getServer(route));
        return send(route, data, channel);
    }

    /**
     * 服务器之间回复消息
     * @param server 目标服务器类型
     * @param data 数据
     * @param replyId 回复的消息 ID
     */
    public static <T> void reply(String server, T data, Long replyId, Channel channel) {
        // 返回消息
        MsgHeader header = new MsgHeader();
        header.setRoute(server + RouteUtils.SERVER_SEPARATOR + "_paradogs.callback");
        header.setReplyMsgId(replyId);
        PRMsg gcMsg = new PRMsg(header, data);
        // 发送 GC
//        Channel channel = getChannel(server);
        channel.writeAndFlush(gcMsg);
        log.debug("[reply] >>>>> {}, {}, {}", server, header, data);
    }

    /**
     * 转发消息: 发送到 Master 服，通过 Master 服转发
     * Gate 服不走这个方法
     * @param byteData 请求头 + byte 数据
     */
    public static <T> void forward(ByteData byteData) {
        PRMsg msg = new PRMsg(byteData.getMsgHeader(), byteData.getDataBytes());
        Channel channel = getChannel(byteData.getMsgHeader().getServer());
        channel.writeAndFlush(msg);
        log.debug("[forward] >>>>> {}, {}", byteData.getMsgHeader().getRoute(), msg.getMsgData());
    }

    /**
     * Gate 服转发消息: 特殊处理，直接发送到对应服
     */
    public static <T> void gateRequest(MsgHeader header, T data, Class<?> returnClass, CompletableFuture promise) {
        PRMsg msg = new PRMsg(header, data);
        String server = RouteUtils.getServer(header.getRoute());
        PRConnection conn = PRConnHolder.getConn(server);
        if (conn == null) {
            throw new MessageException("Not Found Connection, server: " + server);
        }
        // 开启响应监听
        EventUtils.on4Once(header.getRoute(), msg.getMsgHeader().getId(), returnClass, promise);
        conn.getChannel().writeAndFlush(msg);
        log.debug("[request] >>>>> {}, {}", header.getRoute(), msg);
    }

    protected static Channel getChannel(String server) {
        PRConnection conn = PRConnHolder.getConn(serverProperties.isMaster() ? server : "master");
        if (conn == null) {
            throw new MessageException("Not Found Connection, server: " + (serverProperties.isMaster() ? server : "master"));
        }
        return conn.getChannel();
    }

}
